package cn.doitedu.zk.sensor;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.*;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * @date: 2019/7/8
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description:  动态感知系统的服务端程序
 */
public class Provider {

    ZooKeeper zk = null;


    public static void main(String[] args) throws Exception {
        Provider provider = new Provider();

        String hostName = InetAddress.getLocalHost().getHostName();

        // 连接zookeeper注册服务器信息
        provider.connectZookeeper(hostName,Integer.parseInt(args[0]));

        // 做业务去了
        provider.handleService(Integer.parseInt(args[0]));

        // 退出时关闭zk连接
        provider.zk.close();

    }


    public void connectZookeeper(String hostName,int port) throws Exception {

        // 连接zookeeper
        zk = new ZooKeeper("spark01:2181,spark02:2181,spark03:2181", 2000, null);


        // 判断/servers 父节点是否存在
        Stat stat = zk.exists("/servers", false);
        if(stat == null ){
            zk.create("/servers",null, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }

        // 创建zk节点
        String path = zk.create("/servers/server", (hostName + ":" + port).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.err.println(">>>>> 服务器：" + hostName +"上线，在zk上注册了节点： " + path);

    }


    public void handleService(int port) throws Exception {


        ServerSocket ss = new ServerSocket(port);
        while(true){
            Socket sc = ss.accept();
            InputStream in = sc.getInputStream();
            OutputStream out = sc.getOutputStream();

            byte[] buf = new byte[1024];
            int read = in.read(buf);
            System.out.println("收到客户端的请求：" + new String(buf,0,read));

            out.write("你的请求已经处理完成了，这里是结果!".getBytes());
            out.flush();

            in.close();
            out.close();
            sc.close();

        }

    }

}
